/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.tools;

import cn.ac.iie.di.ban.commons.RuntimeEnv;
import cn.ac.iie.di.ban.handler.SE2DBWorker;
import cn.ac.iie.di.ban.metastore.Load2DBEntity;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.apache.log4j.Logger;

/**
 *
 * @author xiaoliang
 */
public class TimeControler implements Runnable {

    private Long lastUpdateTime = -1L;
    private Long currentTime = -1L;
    private InfluxDBHelper dataSzieHelper;
    private InfluxDBHelper recNumHelper;
    private Integer timeout = (Integer) RuntimeEnv.getParam(RuntimeEnv.TIMEOUT);
    private static Logger logger = Logger.getLogger(TimeControler.class.getName());
    private List<Map<String, Object>> insertmsg = new ArrayList<>();
    private BlockingQueue<List<Map<String, Object>>> readBlockingQueue = new ArrayBlockingQueue<>(16);

    public TimeControler(Load2DBEntity load2DBEntity) {
        dataSzieHelper = InfluxDBHelper.getInfluxDBHelper(load2DBEntity.getDb(), load2DBEntity.getDataSizeTable(), load2DBEntity.getIp());
        recNumHelper = InfluxDBHelper.getInfluxDBHelper(load2DBEntity.getDb(), load2DBEntity.getRecNumTable(), load2DBEntity.getIp());
    }

    public boolean addMsg(List<Map<String, Object>> msgs) {
        return readBlockingQueue.offer(msgs);
    }

    public List<Map<String, Object>> getData() {
        try {
            return readBlockingQueue.poll();
        } catch (Exception ex) {
            return null;
        }
    }

    @Override
    public void run() {
        Thread.currentThread().setName("read_insert_influxdb");
        List<Map<String, Object>> msgAll = new ArrayList<>();
        List<Map<String, String>> msgReq = new ArrayList<>();
        lastUpdateTime = System.currentTimeMillis();
        while (true) {
            try {
                List<Map<String, Object>> msg = getData();
                currentTime = System.currentTimeMillis();
                if (msg == null) {
                    try {
                        Thread.sleep(500);
                    } catch (Exception e) {
                    }
                } else {
                    msgAll.addAll(msg);
                    logger.debug(msgAll.size());
                }
                msgAll = dataSizeAndRecNumADD(msgAll);
                if (currentTime - lastUpdateTime >= timeout && !msgAll.isEmpty()) {
                    logger.info("time out and load data forcelly now:" + new Date(currentTime) + " ago:" + new Date(lastUpdateTime) + " queue is:" + readBlockingQueue.size());
                    for (Map<String, Object> msgOne : msgAll) {
                        msgReq.addAll(SE2DBWorker.yclMsg(msgOne));
                    }
                    List<Map<String, String>> recNumMsgs = new ArrayList();
                    List<Map<String, String>> dataSizeMsgs = new ArrayList();
                    SE2DBWorker.shuntmsg(msgReq, recNumMsgs, dataSizeMsgs);
                    recNumHelper.batchInsert(recNumMsgs, "rec_num");
                    dataSzieHelper.batchInsert(dataSizeMsgs, "data_size");
                    lastUpdateTime = System.currentTimeMillis();
                    msgReq.clear();
                    msgAll.clear();
                    Long end_time = System.currentTimeMillis();
                    logger.info("save rec_num " + recNumMsgs.size() + "and data_size " + dataSizeMsgs.size() + " record " + " successfully " + " using " + (end_time - currentTime) + " ms ");
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                msgReq.clear();
                msgAll.clear();
            }
        }
    }

    public List<Map<String, Object>> dataSizeAndRecNumADD(List<Map<String, Object>> docRecords) {
        List<Map<String, Object>> returnmsg = new ArrayList<>();
        Map<String, Long> taskIdRecNumMap = new HashMap<>();
        Map<String, Long> taskIdDataSizeMap = new HashMap<>();
        for (Map<String, Object> record : docRecords) {
            String task_id = record.get("task_id").toString();
            Long data_size = Long.parseLong(record.get("data_size").toString());
            Long tdata_size = taskIdDataSizeMap.get(task_id);
            Long rec_num = Long.parseLong(record.get("rec_num").toString());
            Long trec_num = taskIdRecNumMap.get(task_id);
            if (tdata_size == null) {
                taskIdRecNumMap.put(task_id, rec_num);
                taskIdDataSizeMap.put(task_id, data_size);
                returnmsg.add(record);
            } else {
                taskIdRecNumMap.put(task_id, rec_num + trec_num);
                taskIdDataSizeMap.put(task_id, data_size + tdata_size);
            }
        }
        for (Map<String, Object> record : returnmsg) {
            String task_id = record.get("task_id").toString();
            Long rec_num = taskIdRecNumMap.get(task_id);
            Long data_size = taskIdDataSizeMap.get(task_id);
            record.put("rec_num", rec_num);
            record.put("data_size", data_size);
        }
        return returnmsg;
    }

}
